package com.ku.rocketmq.quickstart.orderconsume;

import com.ku.rocketmq.constants.Const;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @program: rocketmqDemo
 * @description: 顺序消费消息 --- 消费者
 * @author: JokerKu
 * @create: 2020-05-08 10:43
 **/
public class OrderConsumer {

    public OrderConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderConsume_consumer_group");
        consumer.setNamesrvAddr(Const.NAMESERVER_ADDR);
        //设置consumer 第一次启动是从队列头部开始消费还是从队列尾开始消费
        //注意：如果consumer是非第一次启动，那么会按照上次消费的位置继续消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //订阅主题,以及过滤标签内容
        consumer.subscribe("test_order_topic", "TagA");
        //注册监听
        consumer.registerMessageListener(new OrderConsumer.MessageListenerOrderlyImpl());

        consumer.start();
        System.out.println("consumerStart...");
    }

    //MessageListenerOrderly 单线程的监听一个队列
    class MessageListenerOrderlyImpl implements MessageListenerOrderly {

        private final Random random = new Random();

        @Override
        public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext consumeOrderlyContext) {
            for (MessageExt msg : msgs) {
                System.out.println(msg+"----"+new String(msg.getBody()));
                try {
                    //模拟业务逻辑处理
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    //出现异常
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
            }
            return ConsumeOrderlyStatus.SUCCESS;
        }
    }

    public static void main(String[] args) throws MQClientException {
        OrderConsumer consumer = new OrderConsumer();
    }
}
